package com.kevin.data.batch.component;

import com.kevin.data.batch.configuration.BizJobListener;
import jakarta.annotation.Resource;
import org.springframework.batch.core.*;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 读写处理
 */
@Component
public class ReaderWriterComponent {

    @Resource
    private JobRepository jobRepository;

    @Resource
    private BizJobListener jobListener;

    @Resource
    private PlatformTransactionManager transactionManager;

    @Bean("readerWriterJob")
    public Job simpleReaderWriterJob() {
        return new JobBuilder("readerWriterJob", jobRepository)
                .start(readerWriterStep())
                .listener(jobListener)
                .build();
    }

    private Step readerWriterStep() {
        return new StepBuilder("readerWriterStep", jobRepository)
                .<String, String>chunk(3, transactionManager)
                .reader(itemReader()).listener(itemReadListener())
                .processor(itemProcessor()).listener(itemProcessListener())
                .writer(itemWriter()).listener(itemWriteListener())
                .allowStartIfComplete(true)
                .build();
    }

    private ItemReader<String> itemReader() {
        List<String> strings = Arrays.asList("id", "description", "method", "remoteAddr", "userAgent", "clientId");
        Iterator<String> iterator = strings.iterator();
        return () -> iterator.hasNext() ? iterator.next() : null;
    }

    private ItemReadListener<String> itemReadListener() {
        return new ItemReadListener<>() {
            @Override
            public void beforeRead() {
                System.err.println("---------------itemReadListener().beforeRead()--------------------");
            }

            @Override
            public void afterRead(String item) {
                System.err.println("---------------itemReadListener().afterRead(" + item + ")--------------------");
            }

            @Override
            public void onReadError(Exception ex) {
                System.err.println("---------------itemReadListener().onReadError(" + ex.getMessage() + ")--------------------");
            }
        };
    }

    private ItemProcessor<String, String> itemProcessor() {
        return item -> {
            System.err.println("---------------itemProcessor()---Input Data:" + item + "-----------------");
            return item;
        };
    }

    private ItemProcessListener<String, String> itemProcessListener() {
        return new ItemProcessListener<>() {
            @Override
            public void beforeProcess(String item) {
                System.err.println("---------------itemProcessListener().beforeProcess(" + item + ")--------------------");
            }

            @Override
            public void afterProcess(String item, String result) {
                System.err.println("---------------itemProcessListener().afterProcess(" + item + "," + result + ")--------------------");
            }

            @Override
            public void onProcessError(String item, Exception ex) {
                System.err.println("---------------itemProcessListener().onProcessError(" + item + "," + ex.getMessage() + ")--------------------");
            }
        };
    }

    private ItemWriter<String> itemWriter() {
        return (ItemStreamWriter<String>) chunk -> {
            if (!chunk.isEmpty()) {
                for (String str : chunk.getItems()) {
                    System.err.println("---------------itemWriter()---" + str + "-----------------");
                }
            }
        };
    }

    private ItemWriteListener<String> itemWriteListener() {
        return new ItemWriteListener<>() {
            @Override
            public void beforeWrite(Chunk<? extends String> items) {
                System.err.println("---------------itemWriteListener().beforeWrite(" + items.getItems() + ")--------------------");
            }

            @Override
            public void afterWrite(Chunk<? extends String> items) {
                System.err.println("---------------itemWriteListener().afterWrite(" + items.getItems() + ")--------------------");
            }

            @Override
            public void onWriteError(Exception ex, Chunk<? extends String> items) {
                System.err.println("---------------itemWriteListener().onWriteError(" + items.getItems() + "," + ex.getMessage() + ")--------------------");
            }
        };
    }

}
